package com.rabbit.rpc;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * rpc服务端
 */
public class RPCServer {

    // 定义请求队列常量
    private final static String REQUEST_QUEUE_NAME = "rpc_queue";
    // 定义回复队列常量
    private final static String REPLY_QUEUE_NAME = "rpc_reply_queue";

    /**
     * 服务端启动入口
     */
    public static void main(String[] args) {
        // 创建连接
        ConnectionFactory factory = new ConnectionFactory();
        //factory.setHost("localhost");

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 声明请求队列和回复队列
            channel.queueDeclare(REQUEST_QUEUE_NAME, false, false, false, null);
            channel.queueDeclare(REPLY_QUEUE_NAME, false, false, false, null);

            // 每次仅接收一条未经确认的消息
            channel.basicQos(1);

            // 构建消费者属性
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    // 获取消息
                    String message = new String(body, StandardCharsets.UTF_8);

                    // 进行业务处理构建响应数据 (这里做字符串拼接模拟响应数据)
                    String response = message + ":::";

                    // 构造响应基本属性
                    AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                            .correlationId(properties.getCorrelationId())   // 设置关联id
                            .build();

                    // 发送响应数据到回复队列 (rpc_reply_queue)
                    channel.basicPublish("", properties.getReplyTo(), replyProps, response.getBytes(StandardCharsets.UTF_8));

                    // 手动回执消息确认消费
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };

            // 消费rpc_queue队列的消息
            channel.basicConsume(REQUEST_QUEUE_NAME, false, consumer);

            // 持续监听请求消息
            while (true) {
                Thread.sleep(50);
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }

}